package com.bluecc.notifier;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

import java.util.Arrays;

/*
-- Prepare
    ## 增加一个管理员
    $ rabbitmqctl add_user samlet samlet
    $ rabbitmqctl set_user_tags samlet administrator
    $ rabbitmqctl set_permissions -p / samlet ".*" ".*" ".*"
    $ rabbitmqctl list_users
    $ rabbitmqctl list_user_permissions samlet

-- Consume
    $ python -m notifier consume queueName
 */
public class RabbitProcs {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // ...
        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
                .setHost("localhost")
                .setPort(5672)
                .setVirtualHost("/")
                .setUserName("samlet")
                .setPassword("samlet")
                .build();
        DataStream<Integer> integerDataStream = env.fromElements(1, 2, 4, 67, 189)
                .setParallelism(1);
        integerDataStream
                .map(Object::toString)
                .addSink(new RMQSink<String>(
                connectionConfig,            // config for the RabbitMQ connection
                "queueName",                 // name of the RabbitMQ queue to send messages to
                new SimpleStringSchema()));  // serialization schema to turn Java objects to messages

        env.execute();
    }
}
